package rpc

import (
	"encoding/json"
	"errors"
	"fmt"
	"go/ast"
	"io"
	"log"
	"net"
	"net/http"
	"reflect"
	"rpc/codec"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const MagicNumber = 0x3bef5c

//报文结构 --- option | header | body | header | body | .....
//Server

type Option struct {
	MagicNumber int
	CodecType codec.Type
	//连接超时限制
	ConnectionTimeout time.Duration
	//处理超时限制---默认为0--不设限制
	HandlerTimeout time.Duration
}


var DefaultOption = &Option{
	MagicNumber: MagicNumber,
	CodecType: codec.GobType,
	ConnectionTimeout: time.Second*10,
}

// Server Server需要具备的功能
/*
	0.服务每一个请求
	1.读取option设置encoder decoder
	2.读取header
	3.读取body
	4.调用对应函数处理--封装结果
 */
type Server struct {
	serviceMap sync.Map
}

func NewServer() *Server {
	return &Server{}
}

var DefaultServer = NewServer()

func (server *Server) Register(rcvr interface{}) error {
	s := newService(rcvr)
	if _, isLoad := server.serviceMap.LoadOrStore(s.name,s); isLoad {
		return errors.New("rpc: service already defined: "+s.name)
	}
	return nil
}

func Register(rcvr interface{}) error {
	return DefaultServer.Register(rcvr)
}

func (server *Server) findService(serviceMethod string) (svc *service,mType *methodType,err error) {
	dot := strings.LastIndex(serviceMethod,".")
	if dot < 0 {
		err = errors.New("rpc server: service/method request ill-formed: "+serviceMethod)
		return
	}
	serviceName,methodName := serviceMethod[:dot],serviceMethod[dot+1:]
	svci,ok := server.serviceMap.Load(serviceName)
	if !ok {
		err = errors.New("rpc server: can't find service" + serviceName)
		return
	}
	svc = svci.(*service)
	mType = svc.method[methodName]
	if mType == nil {
		err = errors.New("rpc server: can't find method: "+methodName)
	}
	return
}

func (server *Server)Accept(lis net.Listener) {
	for {
		conn,err := lis.Accept()
		if err != nil {
			log.Println("rpc server:accept error:",err)
			return
		}
		go server.ServerConn(conn)
	}
}

func Accept(lis net.Listener)  {
	DefaultServer.Accept(lis)
}

func (server *Server) ServerConn(conn io.ReadWriteCloser)  {
	defer func() {
		err := conn.Close()
		if err != nil {
			return
		}
	}()
	var opt Option
	if err := json.NewDecoder(conn).Decode(&opt); err != nil {
		log.Println("rpc server: option error:",err)
		return
	}
	if opt.MagicNumber != MagicNumber {
		log.Printf("rpc server: invalid maigc number: %x \n",opt.MagicNumber)
		return
	}
	f := codec.NewCodecFuncMap[opt.CodecType]
	if f == nil {
		log.Printf("rpc server: invalid codec type: %s \n",opt.CodecType)
		return
	}
	server.serverCodec(f(conn),&opt)
}

var invalidRequest = struct {}{}

func (server *Server) serverCodec(cc codec.Codec,opt *Option)  {
	sending := new(sync.Mutex)
	wg := new(sync.WaitGroup)
	for {
		req,err := server.readRequest(cc)
		if err != nil {
			if req == nil {
				//说明读取到头
				break
			}
			req.h.Error = err.Error()
			server.sendResponse(cc,req.h,invalidRequest,sending)
			continue
		}
		wg.Add(1)

		go server.handleRequest(cc,req,sending,wg,opt.HandlerTimeout)
	}
	wg.Wait()
	_ = cc.Close()
}

type request struct {
	h *codec.Header
	argv,replyv reflect.Value
	mType *methodType
	svc *service
}

func (server *Server) readRequestHeader(cc codec.Codec) (*codec.Header,error) {
	var h codec.Header
	if err := cc.ReadHeader(&h);err != nil {
		if err!=io.EOF && err!=io.ErrUnexpectedEOF {
			log.Println("rpc server : read header error:",err)
		}
		return nil,err
	}
	return &h,nil
}

func (server *Server) readRequest(cc codec.Codec) (*request,error) {
	h,err := server.readRequestHeader(cc)
	if err!=nil {
		return nil,err
	}
	req := &request{h: h}
	req.svc,req.mType,err = server.findService(h.ServiceMethod)
	if err != nil {
		return req,err
	}
	req.argv = req.mType.newArgv()
	req.replyv = req.mType.newReply()

	//解码需要读取一个ptr来保存数据
	argvi := req.argv.Interface()
	if req.argv.Type().Kind() != reflect.Ptr {
		argvi = req.argv.Addr().Interface()
	}
	if err = cc.ReadBody(argvi); err != nil {
		log.Println("rpc server: read argc error:",err)
		return req,err
	}
	return req,nil
}

func (server *Server) sendResponse(cc codec.Codec,h *codec.Header,body interface{},sending *sync.Mutex)  {
	sending.Lock()
	defer sending.Unlock()
	if err := cc.Write(h,body);err != nil {
		log.Println("rpc server: write response error:",err)
	}
}

func (server *Server) handleRequest(cc codec.Codec,req *request,sending *sync.Mutex,wg *sync.WaitGroup,timeout time.Duration)  {
	defer wg.Done()

	called := make(chan struct{})
	sent := make(chan struct{})

	//存在泄漏的问题---这个协程可能被卡死
	go func() {
		err := req.svc.call(req.mType,req.argv,req.replyv)
		called <- struct{}{}
		if err != nil {
			req.h.Error = err.Error()
			server.sendResponse(cc,req.h,invalidRequest,sending)
			sent <- struct{}{}
			return
		}
		server.sendResponse(cc,req.h,req.replyv.Interface(),sending)
		sent <- struct {}{}

	}()
	if timeout == 0 {
		<- called
		<- sent
		return
	}

	select {
	case <-time.After(timeout):
		req.h.Error = fmt.Sprintf("rpc server handle timeout: expect within %d",timeout)
		server.sendResponse(cc,req.h,invalidRequest,sending)
	case <- called:
		<-sent
	}
}




type methodType struct {
	method reflect.Method
	ArgType reflect.Type
	ReplyType reflect.Type
	numCalls uint64
}

func (m *methodType) NumCalls() uint64 {
	//原子读
	return atomic.LoadUint64(&m.numCalls)
}

func (m *methodType) newArgv() reflect.Value {
	var argv reflect.Value
	//参数可能是指针类型或者值类型
	if m.ArgType.Kind() == reflect.Ptr {
		argv = reflect.New(m.ArgType.Elem())
	} else {
		argv = reflect.New(m.ArgType).Elem()
	}
	return argv
}

func (m *methodType) newReply() reflect.Value {
	//reply一定得是指针类型
	replyv := reflect.New(m.ReplyType.Elem())
	switch m.ReplyType.Elem().Kind() {
	case reflect.Map:
		replyv.Elem().Set(reflect.MakeMap(m.ReplyType.Elem()))
	case reflect.Slice:
		replyv.Elem().Set(reflect.MakeSlice(m.ReplyType.Elem(),0,0))
	}
	return replyv
}

type service struct {
	//结构体名字
	name string
	//结构体类型
	typ reflect.Type
	//结构体本身
	rcvr reflect.Value
	//结构体的方法
	method map[string]*methodType
}

func newService(rcvr interface{}) *service {
	s := new(service)
	s.rcvr = reflect.ValueOf(rcvr)
	s.name = reflect.Indirect(s.rcvr).Type().Name()
	s.typ = reflect.TypeOf(rcvr)
	if !ast.IsExported(s.name) {
		log.Fatalf("rpc server : %s is not valid server name",s.name)
	}
	s.registerMethods()
	return s
}

func (s *service) registerMethods() {
	s.method = make(map[string]*methodType)
	for i := 0;i < s.typ.NumMethod();i++ {
		method := s.typ.Method(i)
		mType := method.Type
		//检验方法的参数是否符合标准
		if mType.NumIn() != 3 || mType.NumOut() != 1 {
			continue
		}
		if mType.Out(0) != reflect.TypeOf((*error)(nil)).Elem() {
			continue
		}
		argType,replyType := mType.In(1),mType.In(2)
		if !isExportedOrBuiltinType(argType) || !isExportedOrBuiltinType(replyType) {
			continue
		}
		s.method[method.Name] = &methodType{
			method: method,
			ArgType: argType,
			ReplyType: replyType,
		}
		log.Printf("rpc server: register %s.%s\n",s.name,method.Name)
	}
}

func isExportedOrBuiltinType(t reflect.Type) bool {
	return ast.IsExported(t.Name()) || t.PkgPath() == ""
}

func (s *service) call(m *methodType,argv,replyv reflect.Value) error {
	atomic.AddUint64(&m.numCalls,1)
	f := m.method.Func
	returnValues := f.Call([]reflect.Value{s.rcvr,argv,replyv})
	if errInter := returnValues[0].Interface(); errInter != nil {
		return errInter.(error)
	}
	return nil
}


const (
	connected = "200 Connected to RPC"
	defaultRPCPath = "/_rpc_"
	defaultDebugPath = "/debug/rpc"
)

func (server *Server) ServeHTTP(w http.ResponseWriter,req *http.Request)  {
	if req.Method != "CONNECT" {
		w.Header().Set("Content-Type","text/plain;charset=utf-8")
		w.WriteHeader(http.StatusMethodNotAllowed)
		_,_ = io.WriteString(w,"405 must CONNECT\n")
		return
	}
	//劫机---截取http连接--为TCP
	conn, _ , err := w.(http.Hijacker).Hijack()
	if err != nil {
		log.Print("rpc hijacking ",req.RemoteAddr,":",err.Error())
		return
	}
	_, _ = io.WriteString(conn,"HTTP/1.0 " + connected + "\n\n")
	server.ServerConn(conn)
}

func (server *Server) HandleHTTP() {
	http.Handle(defaultRPCPath, server)
	http.Handle(defaultDebugPath, debugHTTP{server})
	log.Println("rpc server debug path:", defaultDebugPath)
}

func HandleHTTP()  {
	DefaultServer.HandleHTTP()
}
